﻿using System;
using System.Collections.Generic;
using System.Text;
using NLog;
using System.Threading;
using NGinnBPM.MessageBus;
using NGinnBPM.Lib.Interfaces;
using NGinnBPM.Runtime.Events;
using NGinnBPM.Services;
using NGinnBPM.Runtime.Messages;
using NGinnBPM.MessageBus.Impl;

namespace NGinnBPM.Runtime
{
    /// <summary>
    /// NGinn engine host class. Hosts an instance of NGEnvironment and provides process
    /// execution threads.
    /// Start/stop can be controlled by posting string 'START'/'STOP' commands to messagebus
    /// with subject 'NGinn.Engine.Runtime.NGEngine.Control'.
    /// </summary>
    public class NGEngine : MarshalByRefObject, INGEngine, 
        IMessageConsumer<TaskKickableMessage>, 
        IMessageConsumer<KickTaskEvent>,
        IMessageConsumer<NGinnBPM.Lib.Messages.Heartbeat>,
        IStartableService
    {
        private INGEnvironment _environment;
        private volatile bool _stop = false;
        private Logger log = LogManager.GetCurrentClassLogger();
        private Thread _controllerThread = null;
        private EventWaitHandle _procNotifier = new AutoResetEvent(true);
        private IMessageBus _busEndpoint;
        private Dictionary<string, string> _queuedTasks = new Dictionary<string, string>();
        private ITaskInstanceRepository _instRepository;
        private int _maxConcurrentTasks = 10;

        public NGEngine()
        {
            log.Info("NGEngine created: {0}", this.GetHashCode());
        }
        
        public INGEnvironment Environment
        {
            get { return _environment; }
            set 
            { 
                _environment = value;
            }
        }

        public IServiceResolver ServiceLocator { get; set; }

        public ITaskInstanceRepository TaskInstanceRepository
        {
            get { return _instRepository; }
            set { _instRepository = value; }
        }

        
        public IMessageBus NGinnMessageBus
        {
            get { return _busEndpoint; }
            set
            {
                _busEndpoint = value;
                //_busEndpoint.SubscribeObject(this);
            }
        }

        /// <summary>
        /// Number of tasks that can be 'kicked' simultaneously.
        /// If zero, the scheduling thread will not be started and
        /// the tasks will be 'kicked' by TaskKickableMessages sent to
        /// the message bus.
        /// </summary>
        public int MaxConcurrentTasks
        {
            get { return _maxConcurrentTasks; }
            set { _maxConcurrentTasks = value; }
        }

        public bool ScheduleTasksWithMessageBus { get; set; }

       

        //[MessageBusSubscriber(typeof(string))]
        private void HandleControlMessage(object message)
        {
            string msg = (string)message;
            if (msg == "START")
            {
                Start();
            }
            else if (msg == "STOP")
            {
                Stop();
            }
            else if (msg == "WAKEUP")
            {
                Wakeup();
            }
            else throw new Exception("Unknown command: " + msg);
        }

        private System.Timers.Timer _heartbeatTimer = null;

        public void Start()
        {
            lock (this)
            {
                if (_controllerThread != null) throw new Exception("Engine already started");
                log.Info("Starting....");
                if (Environment == null) throw new Exception("Environment not set");
                _stop = false;
                if (ScheduleTasksWithMessageBus == false)
                {
                    ThreadStart ts = new ThreadStart(ManagerProc);
                    Thread thr = new Thread(ts);
                    thr.Name = "NGinnBPM task scheduler";
                    thr.IsBackground = true;
                    thr.Start();
                    _controllerThread = thr;
                }
                else
                {
                    log.Info("Task scheduling thread not started - message bus handles that");
                }
                _heartbeatTimer = new System.Timers.Timer(10000);
                _heartbeatTimer.AutoReset = true;
                _heartbeatTimer.Elapsed += new System.Timers.ElapsedEventHandler(_heartbeatTimer_Elapsed);
                _heartbeatTimer.Start();
            }
        }

        void _heartbeatTimer_Elapsed(object sender, System.Timers.ElapsedEventArgs e)
        {
            this.NGinnMessageBus.NewMessage(new NGinnBPM.Lib.Messages.Heartbeat())
                .SetNonPersistentLocal(true)
                .Publish();
        }

        public void Stop()
        {
            lock(this)
            {
                log.Debug("Stopping ...");
                if (_heartbeatTimer != null)
                {
                    _heartbeatTimer.Dispose();
                    _heartbeatTimer = null;
                }
                _stop = true;
                if (_controllerThread != null)
                {
                    _controllerThread.Interrupt();
                    if (!_controllerThread.Join(TimeSpan.FromSeconds(30)))
                    {
                        log.Info("Failed to interrupt controller thread, aborting");
                        _controllerThread.Abort();
                    }
                    _controllerThread = null;
                }
                
            }
        }

        /// <summary>
        /// Wakeup - notify the processor thread that something interesting has happened and it should wake up
        /// the processor thread immediately
        /// </summary>
        protected void Wakeup()
        {
            _procNotifier.Set();
        }

        private bool IsTaskEnqueued(string taskId)
        {
            lock(this)
            {
                return _queuedTasks.ContainsKey(taskId);
            }
        }

        private int ProcessingQueueSize
        {
            get
            {
                lock(this)
                {
                    return _queuedTasks.Count;
                }
            }
        }

        private void OnTaskProcessingFinished(string taskId)
        {
            lock (this)
            {
                bool b = _queuedTasks.Remove(taskId);
                if (!b) log.Warn("Task {0} not in queue", taskId);
                if (_queuedTasks.Count == 0)
                    Wakeup();
            }
        }


        private bool EnqueueTask(string taskId)
        {
            lock (this)
            {
                if (_queuedTasks.Count >= MaxConcurrentTasks)
                    return false;
                if (IsTaskEnqueued(taskId)) return false;
                _queuedTasks[taskId] = taskId;
                ThreadPool.QueueUserWorkItem(new WaitCallback(this.KickTask), taskId);
                log.Debug("Task {0} enqueued. Queue size: {1}", taskId, _queuedTasks.Count);
                return true;
            }
        }

        protected bool PollForReadyTasks()
        {
            log.Debug("Querying for ready tasks");
            IList<string> tasx = Environment.GetKickableTasks();
            bool b = false;
            foreach (string taskInstance in tasx)
            {
                if (EnqueueTask(taskInstance))
                    b = true;
            }
            return b;
        }

        protected void ManagerProc()
        {
            try
            {
                while (!_stop)
                {
                    try
                    {
                        if (PollForReadyTasks())
                            _procNotifier.WaitOne(TimeSpan.FromSeconds(1.0), true);
                        else
                            _procNotifier.WaitOne(TimeSpan.FromSeconds(15.0), true);
                    }
                    catch (ThreadInterruptedException) {}
                    catch (Exception ex)
                    {
                        log.Warn("Manager thread error: {0}", ex);
                        DiagnosticEvent de = new DiagnosticEvent("NGEngine", ex);
                        de.Message = "NGEngine scheduler thread error";
                        NGinnMessageBus.Notify(de);
                        if(!_stop) Thread.Sleep(30000);
                    }
                }
            }
            catch (ThreadAbortException) {}
            catch (ThreadInterruptedException) {}
            catch (Exception ex)
            {
                log.Error("Manager thread error, exiting: {0}", ex);
            }
        }

        private void KickTask(object id)
        {
            try
            {
                try
                {
                    KickTaskInternal((string)id, true);
                }
                catch (Exception ex)
                {
                    log.Error("Error kicking task {0}: {1}", id, ex);
                }
                finally
                {
                    OnTaskProcessingFinished((string)id);
                }
            }
            catch (Exception ex)
            {
                log.Error("Catch-all kicking task {0}: {1}", id, ex);
            }
        }

        

        private object KickTaskInternal(string pid, bool handleRetry)
        {
            DateTime dt = DateTime.Now;
            log.Debug("Kicking task {0}", pid);
            try
            {
                Environment.KickTask(pid);
            }
            catch (Exception ex)
            {
                //hm, should be more transactional...
                log.Error("Error kicking process {0}: {1}", pid, ex);
                if (handleRetry)
                {
                    log.Info("Scheduling retry message for process {0}", pid);
                    TaskInstanceRepository.UpdateTaskSchedulingStatus(pid, TaskSchedulingStatus.Error, ex.Message);
                    KickTaskEvent kte = new KickTaskEvent(pid);
                    NGinnMessageBus.Notify(kte);
                }
                else
                {
                    throw;
                }
            }
            finally
            {
                log.Debug("Finished kicking task {0}. Time: {1}", pid, DateTime.Now - dt);
            }
            return null;
        }

        

        /// <summary>
        /// Handle kickprocess retry
        /// </summary>
        /// <returns></returns>
        protected void HandleReadyTaskSaved(TaskKickableMessage msg)
        {
            Wakeup();
        }
        
        /// <summary>
        /// send the NGinn engine heartbeat. To be called periodically
        /// </summary>
        /*public void RunHeartbeat()
        {
            EngineHeartbeat hb = new EngineHeartbeat();
            MessageBus.Notify("NGEngine", "Heartbeat", hb, false);
        }*/

        #region IMessageConsumer<TaskKickableMessage> Members

        public void Handle(TaskKickableMessage message)
        {
            if (ScheduleTasksWithMessageBus)
            {
                Environment.KickTask(message.InstanceId);
            }
            else
            {
                HandleReadyTaskSaved(message);
            }
        }

        #endregion

        #region IMessageConsumer<KickTaskEvent> Members

        public void Handle(KickTaskEvent kpe)
        {
            log.Info("Retrying kick process {0}", kpe.InstanceId);
            try
            {
                KickTaskInternal(kpe.InstanceId, false);
            }
            catch (Exception ex)
            {
                log.Error("RETRY: error kicking process {0}: {1}", kpe.InstanceId, ex);
                throw;
            }
        }

        #endregion

        #region IStartableService Members

        public bool IsRunning
        {
            get { return this._controllerThread != null; }
        }

        #endregion

        public void Handle(Lib.Messages.Heartbeat message)
        {
            log.Trace("heartbeat...");
        }
    }
}

